package com.test;

import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.apache.kafka.clients.consumer.KafkaConsumer;
import org.apache.kafka.clients.consumer.OffsetAndMetadata;
import org.apache.kafka.common.TopicPartition;
import java.util.*;

/**
 * <dl>
 * <dt>Consumer</dt>
 * <dd>Description:消费者</dd>
 * <dd>Copyright: Copyright (C) 2006</dd>
 * <dd>CreateDate: 2019-3-7</dd>
 * </dl>
 *
 * @author Administrator
 */
public class Consumer {

    /**
     * 自动offset提交
     */
    public static void automaticOffsetCommitting(){
        Properties props = new Properties();
        props.put("bootstrap.servers", "192.168.56.101:9092,192.168.56.101:9093");
        props.put("group.id", "test");
        //自动提交
        props.put("enable.auto.commit", "true");
        props.put("auto.commit.interval.ms", "1000");
        props.put("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
        props.put("value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
        KafkaConsumer<String, String> consumer = new KafkaConsumer<String, String>(props);
        // topic
        consumer.subscribe(Arrays.asList("test"));
        //根据topic 和partition来订阅数据
        /*
        TopicPartition topicPartition = new TopicPartition("test", 0);
        consumer.assign(Arrays.asList(topicPartition));
        */
        while (true) {
            //获取数据的超时时间
            ConsumerRecords<String, String> records = consumer.poll(100);
            for (ConsumerRecord<String, String> record : records){
                System.out.printf("offset = %d, key = %s, value = %s%n", record.offset(), record.key(), record.value());
            }
        }
    }
    /**
     * 手工的偏移量控制
     */
    private static void manualOffsetControl(){
        Properties props = new Properties();
        props.put("bootstrap.servers", "192.168.56.101:9092,192.168.56.101:9093");
        props.put("group.id", "test");
        //关闭自动提交
        props.put("enable.auto.commit", "false");
        props.put("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
        props.put("value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
        KafkaConsumer<String, String> consumer = new KafkaConsumer<String, String>(props);
        consumer.subscribe(Arrays.asList("test"));
        final int minBatchSize = 30;
        List<ConsumerRecord<String, String>> buffer = new ArrayList<ConsumerRecord<String, String>>();
        while(true){
            ConsumerRecords<String, String> records = consumer.poll(100);
            for(ConsumerRecord<String, String> record : records){
                buffer.add(record);
                System.out.printf("offset = %d, key = %s, value = %s%n", record.offset(), record.key(), record.value());
            }
            if(buffer.size() >= minBatchSize){
                //insert db
                System.out.println("insert db");
                consumer.commitSync();
                buffer.clear();
            }
        }
    }

    /**
     * 根据不同的分区来进行提交
     */
    private static void handleRecordsInEachPartition(){
        Properties props = new Properties();
        props.put("bootstrap.servers", "192.168.56.101:9092,192.168.56.101:9093");
        props.put("group.id", "test");
        //关闭自动提交
        props.put("enable.auto.commit", "false");
        props.put("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
        props.put("value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
        KafkaConsumer<String, String> consumer = new KafkaConsumer<String, String>(props);
        consumer.subscribe(Arrays.asList("test"));
        try{
            while(true){
                ConsumerRecords<String, String> records = consumer.poll(Long.MAX_VALUE);
                for(TopicPartition partition : records.partitions()){
                    List<ConsumerRecord<String, String>> partitionRecords = records.records(partition);
                    for(ConsumerRecord<String, String> record : partitionRecords){
                        System.out.println(record.offset() + ":" + record.value());
                    }
                    long lastOffset = partitionRecords.get(partitionRecords.size() - 1).offset();
                    consumer.commitSync(Collections.singletonMap(partition, new OffsetAndMetadata(lastOffset + 1)));
                    System.out.printf("提交partition=%s%n", partition);
                }
            }
        }catch(Exception e){
            e.printStackTrace();
        }finally{
            consumer.close();
        }
    }
    public static void main(String[] args) {
        //自动offset提交
//        Consumer.automaticOffsetCommitting();
        // 手工偏移量提交
//        Consumer.manualOffsetControl();
        // 根据不同的分区进行提交
        Consumer.handleRecordsInEachPartition();
    }
}
